package io.xxx.cyb.sync

import com.baomidou.mybatisplus.extension.kotlin.KtQueryWrapper
import io.xxx.cyb.model.*
import io.xxx.cyb.util.dateTimeFormatter
import io.xxx.cyb.util.numberFormat
import org.quartz.*
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.support.BeanDefinitionBuilder
import org.springframework.boot.ApplicationArguments
import org.springframework.boot.ApplicationRunner
import org.springframework.boot.web.client.RestTemplateBuilder
import org.springframework.context.ApplicationContext
import org.springframework.context.support.GenericApplicationContext
import org.springframework.http.client.OkHttp3ClientHttpRequestFactory
import org.springframework.scheduling.quartz.QuartzJobBean
import org.springframework.web.client.RestTemplate
import java.time.Duration
import java.time.LocalDateTime
import java.util.Collections.synchronizedSet

/**
 * 通过门店信息初始化任务。
 * 一个门店对应一个任务，如果门店被禁用/启用则销毁/加载任务。
 */
//@Component
class Synchronizer : ApplicationRunner {

    @Autowired
    private lateinit var storeMapper: StoreMapper

    @Autowired
    private lateinit var platformJobMapper: PlatformJobMapper

    @Autowired
    private lateinit var applicationContext: ApplicationContext

    @Autowired
    private lateinit var scheduler: Scheduler

    override fun run(args: ApplicationArguments) {
        val platformJobs = platformJobMapper.selectList(null)
        for (platformJob in platformJobs) {
            val wrapper = KtQueryWrapper(Store::class.java)
                .eq(Store::pid, "pid")
            val stores = storeMapper.selectList(wrapper)
            for (store in stores) {
                val builder = BeanDefinitionBuilder.genericBeanDefinition(platformJob.jobClass!!)
                    .addConstructorArgValue(store.id)
                val beanDefinition = builder.beanDefinition
                val beanName = beanDefinition.beanClass.simpleName.replaceFirstChar { it.lowercase() } + "Porter"
                (applicationContext as GenericApplicationContext).registerBeanDefinition(beanName, beanDefinition)

                val porter = applicationContext.getBean(beanName, Porter::class.java)
                val jobDetail = JobBuilder.newJob(ProxyJob::class.java)
                    .withIdentity("${beanName}Job", "SYNC")
                    .usingJobData(JobDataMap(mapOf("store" to store, "porter" to porter)))
                    .build()
                val trigger = TriggerBuilder.newTrigger()
                    .withIdentity("${beanName}Trigger", "SYNC")
                    .withSchedule(CronScheduleBuilder.cronSchedule(porter.cron))
                    .build()
                if (!scheduler.checkExists(jobDetail.key)) {
                    scheduler.scheduleJob(jobDetail, trigger)
                    log.info("任务[$beanName]已注册")
                }
            }
        }
    }

    companion object {
        private val log: Logger = LoggerFactory.getLogger(Synchronizer::class.java)
    }
}

@DisallowConcurrentExecution
class ProxyJob : QuartzJobBean() {

    override fun executeInternal(context: JobExecutionContext) {
        val dataMap = context.mergedJobDataMap
        (dataMap["porter"] as Porter).run()
    }
}


/**
 * 数据同步拆分为2部分：
 * 1. 获取时间段内的数据总数
 * 2. 分页获取时间段内的数据
 * 平台响应的数据状态如果为失败，则抛出RuntimeException。
 */
@Suppress("SpringJavaAutowiredMembersInspection")
abstract class Porter(private var storeJob: StoreJob) {

    @Autowired
    private lateinit var applicationContext: ApplicationContext

    @Autowired
    protected lateinit var jobMapper: JobMapper

    @Autowired
    protected lateinit var documentMapper: DocumentMapper

    @Autowired
    protected lateinit var platformMapper: PlatformMapper

    @Autowired
    protected lateinit var platformJobMapper: PlatformJobMapper

    @Autowired
    protected lateinit var storeMapper: StoreMapper

    @Autowired
    protected lateinit var storeJobMapper: StoreJobMapper

    @Autowired
    private lateinit var restTemplateBuilder: RestTemplateBuilder

    protected val restTemplate: RestTemplate by lazy {
        restTemplateBuilder
            .setConnectTimeout(Duration.ofMinutes(1))
            .setReadTimeout(Duration.ofMinutes(10))
            .requestFactory(OkHttp3ClientHttpRequestFactory::class.java)
            .build()
    }

    /**
     * 计划任务
     */
    open val cron = "0/1 * * * * ?"

    /**
     * 限制开始时间和结束时间的时间范围不超过平台的限制
     */
    protected open val duration: Duration = Duration.ZERO

    /**
     * 数据延迟时长
     */
    protected open val delay: Duration = Duration.ZERO

    /**
     * 使用不同的参数多次调用接口
     * @return 第一个参数为扩展参数，第二个参数为并行标志
     */
    protected open fun acquire() {
    }

    /**
     * 申请可继续操作的资源，ex：进行限流等操作
     */
    protected open fun getParameters(store: Store): Pair<List<Any?>, Boolean> = listOf(null) to false

    /**
     * 返回平台限制的开始时间，部分平台只允许查询3个月以内的订单
     */
    protected open fun getStartTime(storeJob: StoreJob): LocalDateTime = storeJob.endTime!!

    /**
     * 分页起始页
     */
    protected open val startPage = 1

    /**
     * 每页大小
     */
    protected open val pageSize = 100

    fun run() {
        fun getStore(): Store? {
            val store = storeMapper.selectById(storeJob.sid)
            val platform = platformMapper.selectById(store.pid)
            store.platform = platform
            return store
        }

        /**
         * 检查任务是否需要销毁或暂停
         * @return 0 - 任务正常，1 - 任务暂停，2 - 任务销毁
         */
        fun validJob(): Int {
            return 0
        }

        fun getProperty(storeJob: StoreJob, propName: String): String? {
            return storeJob.props.getString(propName) ?: storeJob.platformJob!!.props.getString(propName)
        }

        fun getDelay(storeJob: StoreJob): Duration {
            val delayProp = getProperty(storeJob, "delay")
            return if (delayProp == null) this.delay else Duration.ofSeconds(delayProp.toLong())
        }

        /**
         * 数据同步需要一对开始时间和结束时间，根据平台对于开始时间和结束时间的时间范围约束生成
         */
        fun getTimeRanges(startTime: LocalDateTime, endTime: LocalDateTime): Set<Pair<LocalDateTime, LocalDateTime>> {
            val timeRanges = hashSetOf<Pair<LocalDateTime, LocalDateTime>>()
            if (duration == Duration.ZERO) {
                timeRanges.add(startTime to endTime)
            } else {
                val totalSeconds = Duration.between(startTime, endTime).abs().seconds
                val indexes = totalSeconds / duration.seconds + if (totalSeconds % duration.seconds == 0L) 0 else 1
                for (i in 0 until indexes) {
                    val first = startTime.plusSeconds(duration.seconds * i)
                    val second = if ((i + 1) == indexes) endTime
                    else startTime.plusSeconds(duration.seconds * (i + 1))
                    timeRanges.add(first to second)
                }
            }
            return timeRanges
        }

        val start = System.currentTimeMillis()

        val store = getStore() ?: return
        when (validJob()) {
            1 -> return
            2 -> applicationContext.publishEvent(JobDestroyEvent(storeJob))
        }
        val sjWrapper = KtQueryWrapper(StoreJob::class.java)
            .eq(StoreJob::sid, store.id)
            .eq(StoreJob::jid, storeJob.jid)
        storeJob = storeJobMapper.selectOne(sjWrapper)
        if (!storeJob.enabled!!) return
        if (storeJob.endTime == null) {
            storeJob.endTime = store.created
            storeJobMapper.updateById(storeJob)
        }
        storeJob.store = store
        storeJob.job = jobMapper.selectById(storeJob.jid)
        val pjWrapper = KtQueryWrapper(PlatformJob::class.java)
            .eq(PlatformJob::pid, store.pid)
            .eq(PlatformJob::jid, storeJob.jid)
        storeJob.platformJob = platformJobMapper.selectOne(pjWrapper)

        val startTime = getStartTime(storeJob)
        val endTime = LocalDateTime.now().minusMinutes(getDelay(storeJob).toMinutes())
        if (startTime >= endTime) return
        val timeRanges = getTimeRanges(startTime, endTime)
        val result = synchronizedSet(hashSetOf<String>())
        val (parameters, parallel) = getParameters(store)
        val stream = parameters.stream()
        if (parallel) stream.parallel()
        stream.forEach {
            for (timeRange in timeRanges) {
                acquire()
                result.addAll(pullAndSave(store, timeRange.first, timeRange.second, it))
            }
        }
        val spend = numberFormat.format((System.currentTimeMillis() - start) / 1000.0)
        val st = startTime.format(dateTimeFormatter)
        val et = endTime.format(dateTimeFormatter)
        log.info("${javaClass.simpleName}(${store.id})[$st - $et]同步 ${result.size} 条数据完成，耗时 $spend 秒")
    }

    /**
     * 拉取并保存数据
     * @return 单号列表
     */
    protected open fun pullAndSave(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): List<String> {
        val count = getCount(store, startTime, endTime, parameter) ?: return emptyList()
        var pages = count / pageSize + if (count % pageSize > 0) 1 else 0
        val result = arrayListOf<String>()
        while (pages-- > 0) {
            val page = pages + startPage
            val data = getData(store, startTime, endTime, parameter, page)
            result.addAll(save(store, data))
        }
        return result
    }

    /**
     * 返回时间范围内的数据总数。
     * @param store 店铺信息
     * @param startTime 开始时间
     * @param endTime 结束时间
     * @param parameter 扩展参数
     * @return 数据总数
     */
    protected open fun getCount(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?
    ): Long? = throw NotImplementedError()


    /**
     * 返回时间范围内的文档列表。
     * @param store 店铺信息
     * @param startTime 开始时间
     * @param endTime 结束时间
     * @param parameter 扩展参数
     * @return 文档列表
     */
    open fun getData(
        store: Store,
        startTime: LocalDateTime,
        endTime: LocalDateTime,
        parameter: Any?,
        pageNo: Long
    ): List<Document>? = throw NotImplementedError()

    /**
     * 保存不存在的文档，更新已存在的文档（更新时间大于已存在的文档）
     * @return 文档外部编码
     */
    protected open fun save(store: Store, data: List<Document>?): List<String> {
        if (data.isNullOrEmpty())
            return emptyList()
        data.forEach { if (it.sid == null) it.sid = store.id }
        documentMapper.upsertAll(data)
        return data.map { it.sn!! }
    }

    companion object {
        private val log: Logger = LoggerFactory.getLogger(Porter::class.java)

        /**
         * 记录错误日志并抛出RuntimeException。
         * @param className Porter类简名
         * @param api url path 或 method
         * @param code 错误码
         * @param msg 错误消息
         */
        fun throwException(className: String, api: String?, code: Any?, msg: Any?) {
            val message = String.format("%s: %s call failed, %s[%s]", className, api, code, msg.toString())
            log.error(message)
            throw RuntimeException(message)
        }
    }
}